package org.tigr.htc.server;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.File;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.NoSuchElementException;

import org.apache.log4j.Logger;
import org.tigr.antware.shared.command.events.CommandRuntimeStatusLI;
import org.tigr.antware.shared.command.events.CommandStatusLI;
import org.tigr.antware.shared.command.events.FailureEvent;
import org.tigr.antware.shared.command.events.FinishEvent;
import org.tigr.antware.shared.command.events.InterruptEvent;
import org.tigr.antware.shared.command.events.ResumeEvent;
import org.tigr.antware.shared.command.events.StartEvent;
import org.tigr.antware.shared.command.events.SubmitEvent;
import org.tigr.antware.shared.command.events.SuspendEvent;
import org.tigr.htc.cmd.CmdState;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.common.ConfigUpdateLI;
import org.tigr.htc.common.HTCConfig;

/**
 * The <b>Scheduler</b> class queues commands waiting to be submitted to the 
 * grid.
 */    
public class Scheduler implements Runnable, CommandRuntimeStatusLI, CommandStatusLI, ConfigUpdateLI {
    static private Logger logger = Logger.getLogger(Scheduler.class);
    final static String submitMessage = "Command Submitted to htcserver version 1.2";
    private static Scheduler sched = null;
    private static File cwd = new File(HTCConfig.getProperty("CWD"));
    boolean acceptNew = true;
    boolean holdSubmits = false;
    int runningCount = 0;
    int pendingCount = 0;
    public static int MASK = 0x3fff;
    int MAXRUNNING = Integer.parseInt(HTCConfig.getProperty("MAXRUNNING"));
    
    /**
     * The variable <code>fifoSched</code> holds the schduling flag based on the type of scheduling 
     * specified in config default is FIFO scheduling which is the simplest
     */
    private static boolean fifoSched = HTCConfig.getProperty("SCHEDULING").equals("RoundRobin") ? false : true;
    
    /**
     * The variable <code>fifoWaitQ</code> holds waiting commands for FIFO implementation. If the server is configured 
     * to use FIFO then the first command added to the stack is submitted
     */
    private LinkedList fifoWaitQ;
    
    // 
    /**
     * The variable <code>roundRobinQ</code> holds waiting user queues for round robin user based scheduling
     * During scheduling the first request from the next user is scheduled.
     */
    private LinkedList roundRobinQ;
    /**
     * The variable <code>rrIndex</code> holds current index within the round robin queue
     */
    private int rrIndex = 0;
    /**
     * The variable <code>userQueueMap</code> holds a mapping of user names to user queues to make it 
     * easier to detect if a queue for this user already exists
     */
    private HashMap userQueueMap;
    
    // all runners
    private HashMap runners;
    
    // running commands
    private HashMap activeq;

    private Scheduler() throws RuntimeException {
        fifoWaitQ = new LinkedList();
        roundRobinQ = new LinkedList();
        userQueueMap = new HashMap();
        activeq = new HashMap();
        runners = new HashMap();
        if(cwd.exists()) {
            logger.warn(" CWD exists, no need to make working directory " + cwd);
        } else {
            logger.debug("making cwd " + cwd);
            if(!cwd.mkdirs()) {
                logger.fatal(" Couldn't create working directory needed by server " + cwd);
                throw new RuntimeException("Config error, cwd could not be created");
            }
        }
    }
    
    /**
     * The method <code>getParentDir</code> returns the parent directory for the command with
     * the specified ID where all the command files are stored
     * 
     * @param id a <code>long</code>
     * @return a <code></code>
     */
    public static File getParentDir(long id) {
        File parentDir = new File(cwd, Long.toString(id & ~MASK));
        return parentDir;
    }
    
    /**
     * The method <code>setAcceptNew</code> instructs the server to start or stop accepting 
     * new requests.
     * 
     * @param flag a <code>boolean</code> of true to indicate that the server should accept
     * new requests, false to stop accepting new requests
     */
    public synchronized void setAcceptNew(boolean flag) { 
        if (logger.isDebugEnabled()) {
            logger.debug(" setting accept new flag to " + flag);
        }
        acceptNew = flag; 
    }
    
    /**
     * The method <code>isAcceptNew</code> returns the status of the accept new flag. This flag 
     * indicates if the server is accepting new requests.
     * 
     * @return a <code>boolean</code> of true if new requests are being accepted
     */
    public synchronized boolean isAcceptNew () {
        return acceptNew;
    }
    
    /**
     * The method <code>setHoldSubmits</code> instructs the server to start or stop submitting 
     * requests to the grid.
     * 
     * @param flag a <code>boolean</code> of true to hold submitting requests, false to resume
     * submitting requests
     */
    public synchronized void setHoldSubmits(boolean flag) {
        if (logger.isDebugEnabled()) {
            logger.debug(" setting hold submit flag to " + flag);
        }
        holdSubmits = flag;
        
        // If the hold on submits is removed then notify potential waiting thread to 
        // start submitting
        if (!flag) {
            // Send a notification to all waiting threads that hold has been removed
            this.notifyAll();
        }
    }
    
    /**
     * The method <code>isHoldSubmits</code> returns the status of the hold submits flag
     * 
     * @return a <code>boolean</code> of true if the submits are held, false otherwise
     */
    public synchronized boolean isHoldSubmits () {
        return holdSubmits;
    }
    
    /**
     * The method <code>getInstance</code> method returns the singleton instance of the scheduler.
     * 
     * @return
     */
    public synchronized static Scheduler getInstance() {
        if (sched == null) {
            sched = new Scheduler();
        }
        return sched;
    }
    
    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    public void run() {
        Thread.currentThread().setName("Main Scheduler ");
        HTCConfig.addConfigUpdateListener(this);
        schedule();
        HTCConfig.removeConfigUpdateListener(this);
    }
    
    /**
     * The method <code>schedule</code> is the main method that submits the queued requests to the
     * underlying grid. The method sleeps if there are no pending jobs, or if the maximum running
     * jobs limit has been reached, or if the submits have been put on hold.
     * <p>
     * Else the method picks up the next queued command and launches the runner to chaperon the 
     * execution of the command.
     * 
     */
    public void schedule() {
        IRunner runner;
        while (true) {
            // check wait q
            try {
                synchronized (this) {
                    while((pendingCount == 0) ||
						  (runningCount >= MAXRUNNING) || 
						  holdSubmits) {
                        if (logger.isDebugEnabled()) {
                            logger.debug(" wait for a new commands or some to finished");
                            logger.debug(" jobs running " + runningCount);
                            logger.debug(" jobs waiting " + pendingCount);
                            logger.debug(" Maxrunning is set to " + MAXRUNNING);
                        }
                        this.wait();
                        logger.debug(" wake and check queue");
                    }
					// before attempting to remove the runner make sure that there are runners
					// as the thread may have gone into wait because of a hold and been awakened
					// by releasing the hold and not because of new commands were added
					if (pendingCount == 0) {
						continue;
					}
                    
                    // Based on the type of scheduling submit the request to the grid
                    if (fifoSched) {
                        runner = (IRunner) fifoWaitQ.removeFirst(); // Get the first pending runner
                    }
                    else {
						if (logger.isDebugEnabled()) {
                            logger.debug("  current round robin index  " + rrIndex);
                        }

                        // Else get the user list for the next user and get the first 
                        // runner off this list
                        LinkedList userQ = (LinkedList) roundRobinQ.get(rrIndex);
                        runner = (IRunner) userQ.removeFirst(); // Get the first pending runner for current user                          
                        String userName = ((Command) runner.getICommand()).getUserName();
                        
                        if (logger.isDebugEnabled()) {
                            logger.debug("  submitting request for user " + userName);
                        }
                        
                        // If at the end of this there are no more pending commands for this user
                        // remove the queue entry for this user from the rrQ and the user queue map
                        if (userQ.size() == 0) {
                            roundRobinQ.remove(rrIndex);
                            userQueueMap.remove(userName);
                            userQ = null;
							rrIndex--; // Decrement the rr index to adjust for the entry for this user being removed
                        }
                        
                        // If we have reached the end of the round robin Q then restart the index
                        if (++rrIndex >= roundRobinQ.size()) {
                            rrIndex = 0;
                        }
                    }
                }
                
                // Launch the thread to process the command
                startRunner(runner);
                    
                // Increase the running count and decrement the pending count
                runningCount++;
                pendingCount--;
                if (logger.isDebugEnabled()) {
                    logger.debug(" running count " + runningCount + 
                            " pending count " + pendingCount);
                }
                
            } catch (NoSuchElementException e) {
                logger.warn(" first element not there yet size is " + fifoWaitQ.size());
            } catch (InterruptedException e) {
                logger.fatal(" scheduler was interrupted ");
                return;
            }
        }
        
    }
    
    /**
     * The method <code>startRunner</code> method starts the execution of the runner.
     * A new thread is created to process the runner
     * 
     * @param runner
     */
    public void startRunner(IRunner runner) {
        long id = runner.getICommand().getID();
        
        activeq.put(new Long(id), runner);
        Thread t = new Thread(runner, "Command " + id);
        logger.debug("Starting runner for Command ID: " + id);
        t.start();
    }
    
    /**
     * The method <code>remonitorCommand</code> method creates and adds a runner to monitor the
     * command passed as a parameter. This method should be called in situations where the command was 
     * launched and the server crashed so we can create another runner to monitor for changes
     * in the execution status for this command. Based on the last known status of the command
     * the runner is either added to the waiting queue or active jobs queue.
     * 
     * @param cmd a <code>ICommand</code> representing the command that should be re-monitored 
     * @throws Exception
     */
    public synchronized void remonitorCommand (ICommand cmd) throws Exception {
        // Create a new runner and add this runner to the ones monitored by
        // the scheduler
        long id = cmd.getID();
        Long ID = new Long(id);
        File parentDir = Scheduler.getParentDir(id);
        if (logger.isDebugEnabled()) {
            logger.debug(" create runner for command with ID: " + ID);
        }
        
        // Create a runner and add based on the state of the command either add to 
        // active runners or queue the runner
        IRunner runner = RunnerFactory.newRunner(cmd, parentDir);
        if ((cmd.getState() == CmdState.INIT)) {
            if (logger.isDebugEnabled()) {
                logger.debug("  queuing command with ID: " + ID);
            }
            // As the command is in init state queue the command
            addToQueue(runner);
        }
        else {
            // As the command was already waiting or running on the grid
            // add it to the active queue and register it for monitoring
            if (logger.isDebugEnabled()) {
                logger.debug("  remonitoring command with ID: " + ID);
            }
            // As the command was either waiting or running on the grid
            // just add runner to monitor queue
            runners.put(ID, runner);
            activeq.put(ID, runner);
            runningCount++; // Increment the running count

            // Register the runner with the monitor
            RunnerMonitor.register(runner); 
            
            // Add this as the class to get events from the command
            cmd.addRuntimeStatusListener(this);
            cmd.addStatusListener(this);  
        }     
    }

    
    /**
     * The method <code>enqueue</code> adds the command to the scheduling queue. This method
     * builds a runner that will chaperone this command. The method creates the directory
     * that will hold all the command related files such as the submit and event files among others.
     * <p>
     * It will then add the runner to the waiting queue to be picked up by the scheduler in its 
     * next scheduling cycle.
     * 
     * @param pCmd a <code>ICommand</code> that needs to be executed on the grid
     * @throws Exception
     */
    public void enqueue(ICommand pCmd) throws Exception {
        // If the server is accepting connections then process, otherwise throw an
        // exception
        if (acceptNew) {
            try {
                logger.debug("Enter enqueue cmd id: " + pCmd.getID());
                logger.debug("Notify scheduler of new command");
                pCmd.createCommand();
                
                
                IRunner runner = null;
                long lid = pCmd.getID();
                try {
                    // Create the parent directory of the command directory that will hold the 
                    // command files. The parent directory is created by masking off the lower order 
                    // bytes of the ID. The applied mask is 0x3fff  
                    logger.debug("Starting command " + lid);
                    logger.debug("Command id masks to " + (lid & ~MASK));
                    File parentDir = new File(cwd, Long.toString(lid & ~MASK));
                    
                    if (!parentDir.exists()) {
                        logger.debug("Making parent directory: " + parentDir);
                        if (!parentDir.mkdirs()) {
                            logger
                            .fatal("Couldn't create directory needed by server "
                                    + parentDir);
                            throw new Exception("Config error, " + parentDir
                                    + "could not be created");
                        }
                    }
                    
                    // Create the runner for this Command
                    logger.debug("Create runner.");
                    runner = RunnerFactory.newRunner(pCmd, parentDir);
                } catch (Exception e) {
                    logger.error("Exception in scheduler  " + e.getMessage());
                    // fire failed event
                    pCmd.fireFailureEvent(e);
                    pCmd.updateCommand();
                }
                
                // Add this command and runner to the queue
                addToQueue(runner);
                
            } catch (Exception e) {
                logger.warn("scheduler" + e.getMessage(), e);
                // fire failed event
                pCmd.fireFailureEvent(e);
                throw e;
            }
        } else {
            throw new Exception(
            " Server currently not accepting new request submissions");
        }       
    }

    /**
     * The method <code>addToQueue</code> queues the runner based on the current
     * scheduling mechanism.
     * 
     * @param runner <code>IRunner</code> that chaperones the execution of the user command
     */
    private void addToQueue(IRunner runner) {
        if (logger.isDebugEnabled()) {
            logger.debug("Adding runner for Command with ID: " + runner.getICommand().getID());
        }
        Command cmd = (Command) runner.getICommand();
        long lid = cmd.getID();
        
        // Add the runner to the waiting queue and register the scheduler
        // as the runtime and status listener to receive events from the command
        synchronized (this) {
            if (fifoSched) {
                fifoWaitQ.add(runner); // Add the runner to the waiting queue
            }
            else {
                LinkedList userQ = null;
                String userName = cmd.getUserName();
                
                // As the round robin scheduling has been specified use the user name
                // to create an entry in the q
                if (userQueueMap.containsKey(userName)) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Reusing existing list for user " + userName);
                    }
                    // If an entry already exists for this user then 
                    // get the linked list for this user and pick up the first 
                    // command submitted by this user
                    userQ = (LinkedList) userQueueMap.get(userName);
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("  creating new list for user " + userName);
                    }
                    
                    // As an entry does not exist then create a new queue for this user
                    // and add to the end of the round robin queue as well as to the user queue 
                    // map
                    userQ = new LinkedList();
                    userQueueMap.put(userName, userQ);
                    roundRobinQ.addLast(userQ);
                }
                userQ.add(runner); // Add the runner at the end of the list for this user
            }
            pendingCount++; // Increment the pending count
            runners.put(new Long(lid), runner); // Add the runner to the runners map
            
            // Register this as the listener to receive command events
            cmd.addRuntimeStatusListener(this);
            cmd.addStatusListener(this);
            
            // Send a notification to all waiting threads that a new runner has
            // been added
            this.notifyAll();
        }
    }
    
    /**
     * The method <code>findRunner</code> locates the runner responsible for chaperoning
     * the command with the specified ID
     * 
     * @param id a <code>long</code> representing the command ID
     * 
     * @return a <code>IRunner</code> for this command, null if a matching runner does not
     * exist
     */
    public IRunner findRunner(long id) {
        return (IRunner) runners.get(new Long(id));
    }
    
    /**
     * The method <code>dequeue</code> removes the command with the specified ID from the
     * currently active commands being processed by the server
     * 
     * @param id a <code>long</code> representing the command ID to dequeue
     */
    public void dequeue(long id) {
        logger.debug("Dequeue command " + id);
        Long lid = new Long(id);
        try {
            synchronized (this) {
                runningCount--; // Reduce the running count
                activeq.remove(lid);  // remove from active queue
                runners.remove(lid); // remove the runner from runners queue
                this.notifyAll(); // send notification that a command has been dequeued
            }
            
            logger.debug("Decrease Commands running " + runningCount);
        } catch (Exception e) {
            logger.debug(e.getMessage(), e);
        }
    }

    /**
     * The method <code>getPendingCount</code> returns the value for the field pendingCount. 
     * This is the total number of jobs queued in the HTC service.
     *
     * @return a <code>int</code> representing the value for the field <code>pendingCount</code>.
     * 
     */
    public synchronized int getPendingCount() {
        return this.pendingCount;
    }

    /**
     * The method <code>getRunningCount</code> returns the value for the field runningCount.
     * 
     * @return an <code>int</code> representing the current count of active runners which include
     * jobs running and waiting in the grid.
     */
    public synchronized int getRunningCount() {
        return this.runningCount;
    }
    
    /**
     * The method <code>getActiveRunnerCount</code> returns the number of runners that have jobs running
     * or waiting on the grid. Typically this count should match the one returned by <code>getRunningCount</code>
     * but in case there is an error there may be divergence and this method would allow the admin to 
     * detect this discrepancy.
     * 
     * @return an <code>int</code>
     */
    public synchronized int getActiveRunnerCount() {
        return activeq.size();
    }
    
    /**
     * The method <code>getPendingRunnerCount</code> returns the number of runners in the pending
     * queue. Typically this count should match the pending count but just in case there is some
     * error this method would allow the admin to check them
     * 
     * @return an <code>int</code> representing the number of waiting runners in the queue.
     */
    public synchronized int getPendingRunnerCount() {
        // Based on the type of scheduling compute the
        // pending running count
        int runnerCount = 0;
        if (fifoSched) {
            // As this is FIFO, scheduling the count is the size of the queue
            runnerCount = fifoWaitQ.size();
        }
        else {
            // As this is round robin, scheduling loop through the
            // entire queue to compute the total pending count
            for (Iterator iter = roundRobinQ.iterator(); iter.hasNext();) {
                runnerCount += ((LinkedList) iter.next()).size();
            }
        }
        return runnerCount;
    }
    
    /**
     * The method <code>setFIFOScheduling</code> sets the current scheduling policy
     * 
     * @param pFlag a <code></code>
     */
    public void setFIFOScheduling(boolean pFlag) {
		fifoSched = pFlag;
	}

    
    // listener stuff
    
    // Do nothing on command submit event
    public void commandSubmitted(SubmitEvent event) {
        logger.debug("Got :" + event.toString());
        logger.debug("Submitter was " + event.getUser());
    }
    
    // Do nothing on start event
    public void commandStarted(StartEvent event) {
        logger.debug("Got :" + event.toString());
        logger.debug("Execute location was " + event.getMachine());
    }
    
    // Dequeue the Command on finish event
    public void commandFinished(FinishEvent event) {
        logger.debug("Got :" + event.toString());
        logger.debug("Return value was " + event.getReturnValue());
        dequeue(event.getCommandID());
    }
    
    // Dequeue the Command on failure
    public void commandFailed(FailureEvent event) {
        logger.debug("Got :" + event.toString());
        logger.debug("failure was " + event.getMessage());
        dequeue(event.getCommandID());
    }
    
    // Dequeue the Command on interruption
    public void commandInterrupted(InterruptEvent event) {
        logger.debug("Got :" + event.toString());
        dequeue(event.getCommandID());
    }
    
    public void commandSuspended(SuspendEvent event) {
        logger.debug("Got :" + event.toString());
    }
    
    public void commandRestarted(StartEvent event) {
        logger.debug("Got :" + event.toString());
        logger.debug("Execute location was " + event.getMachine());
    }
    
    public void commandResumed(ResumeEvent event) {
        logger.debug("Got :" + event.toString());
    }
    
    public void configUpdated() {
        logger.debug("Scheduler got a new config update... Reloading.");
        cwd = new File(HTCConfig.getProperty("CWD"));
        MAXRUNNING = Integer.parseInt(HTCConfig.getProperty("MAXRUNNING"));
    }   
}
